package queueasync;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.IntStream;

/**
 * 任务管理
 *
 * @author dingwen
 */
public class TaskManagement {
    public static ConcurrentHashMap<Integer,Object> resultMap = new ConcurrentHashMap<>();
    /**
     * 某场景下 排队办事的执行体
     */
    private final BlockingQueue<Task> taskQueue;

    /**
     * 业务执行线程
     */
    public List<TaskExecutorThread> taskExecutorThreads;

    /**
     * 任务执行人线程大小
     */
    private final int taskExecutorThreadsSize;


    /**
     * 初始化，任务队列，执行线程
     *
     * @param taskExecutorThreadsSize 任务执行人线程大小
     */
    public TaskManagement(int taskExecutorThreadsSize) {
        this.taskExecutorThreadsSize = taskExecutorThreadsSize;
        taskQueue = new LinkedBlockingQueue<>();
        taskExecutorThreads = new ArrayList<>(taskExecutorThreadsSize);
    }

    /**
     * 场景开始启动
     */
    public void start() {
        //防止存在未停止的业务线程，停止所有业务执行线程
        stop();

        //启动业务执行线程处理任务
        IntStream.range(0, taskExecutorThreadsSize).forEach(index -> {
            // 为每一个业务执行线程分配任务队列
            TaskExecutorThread taskExecutorThread = new TaskExecutorThread(taskQueue, index);
            taskExecutorThreads.add(taskExecutorThread);
            // 启动任务执行线程
            taskExecutorThread.start();
        });

    }

    /**
     * 停止所有业务执行线程
     */
    public void stop() {
        Optional.ofNullable(taskExecutorThreads).ifPresent((t) -> {
            taskExecutorThreads.stream().parallel().forEach(TaskExecutorThread::quit);
        });
    }

    /**
     * 添加任务队列
     *
     * @param task 任务
     */
    public <T extends Task> int add(Task task) {
        if (!taskQueue.contains((task))) {
            taskQueue.add(task);
        }
        return task.hashCode();
    }

}

